package trouger.dmt;

import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashSet;

import javax.xml.soap.Text;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;

/**
 * The root base class of all Runner classes.
 * Create instance of JoinTableRunner, SelfAggregationRunner, or DumpTableRunner, to do corresponding operations.
 *
 * @param <M> a mapper class that extends the generic BaseMapper class
 * @param <R> a reducer class that extends the generic BaseReducer class
 */
public abstract class BaseMapredRunner<M, R> {
	
	protected ArrayList<Path[]> inputPathsList = new ArrayList<Path[]>();
	@SuppressWarnings("unchecked")
	protected ArrayList<Class<? extends FileInputFormat> > inputFormatClassesList = new ArrayList<Class<? extends FileInputFormat> >();
	protected ArrayList<String> pathPatternList = new ArrayList<String>();
	protected ArrayList<String> charsetList = new ArrayList<String>();
	protected ArrayList<String> separatorList = new ArrayList<String>();
	protected ArrayList<Integer> segCountList = new ArrayList<Integer>(); 
	protected int maxReduceItemCount = 0;
	protected Path outputPath = null;
	protected String outputSeparator = null;
	protected String nullValueNotation = null;
	
	protected String[] keyColumnList = null;
	protected String[] valueColumnList = null;
	protected String resultFieldsIndices = null;
	
	protected Class<? extends SourceRecordFilter> sourceRecordFilterClass = null;
	protected Class<? extends OutputRecordFilter> outputRecordFilterClass = null;
	protected Class<? extends RecordGroupFilter> recordGroupFilterClass = null;
	
	protected int numMapTasks = 0;
	protected int numReduceTasks = 0;
	protected int numTasksPerJvm = 0;
	
	protected JobConfigurator jobConfigurator = null;
	
	public void setMaxReduceItemCount(int maxCount) {
		maxReduceItemCount = maxCount;
	}
	/**
	 * Add a source table for the current operation.
	 * @param path the file system path of the source table.
	 * @param charset the character set of the source table, e.g. "UTF-8".
	 * @param separator the separator of the source table, a plain text string, no any wildcard or regular expression supported.
	 * @param segCount the expected minimal column count of the source table. If it is 0, that means any number of columns is valid.
	 */
	public void addSourceTable(String path, String charset, String separator, int segCount){
		addSourceTable(new String[]{path}, TextInputFormat.class, path, charset, separator, segCount);
	}
	
	@SuppressWarnings("unchecked")
	public void addSourceTable(String path, Class<? extends FileInputFormat> inputFormat, String charset, String separator, int segCount){
		addSourceTable(new String[]{path}, inputFormat, path, charset, separator, segCount);
	}
	/**
	 * Add a source table for the current operation.
	 * Calling this method is equivalent to calling addSourceTable(path, charset, separator, 0).
	 * @param path the file system path of the source table.
	 * @param charset the character set of the source table, e.g. "UTF-8".
	 * @param separator the separator of the source table, a plain text string, no any wildcard or regular expression supported.
	 */
	public void addSourceTable(String path, String charset, String separator){
		addSourceTable(new String[]{path}, TextInputFormat.class, path, charset, separator, 0);
	}
	
	@SuppressWarnings("unchecked")
	public void addSourceTable(String path, Class<? extends FileInputFormat> inputFormat, String charset, String separator){
		addSourceTable(new String[]{path}, inputFormat, path, charset, separator, 0);
	}
	
	/*public void addSourceTable(String path, String charset){
		addSourceTable(new String[]{path}, path, charset, null, 0);
	}
	public void addSourceTable(String path){
		addSourceTable(new String[]{path}, path, null, null, 0);
	}*/
	
	/**
	 * Add a source table for the current operation.
	 * @param paths all paths that the source table consists of.
	 * @param pathPattern the longest common prefix or longest common substring of paths, or any string that fits. See the manual.
	 * @param charset the character set of the source table, e.g. "UTF-8".
	 * @param separator the separator of the source table, a plain text string, no any wildcard or regular expression supported.
	 * @param segCount the expected minimal column count of the source table. If it is 0, that means any number of columns is valid.
	 */
	@SuppressWarnings("unchecked")
	public void addSourceTable(String[] paths, Class<? extends FileInputFormat> inputFormat, String pathPattern, String charset, String separator, int segCount){
		Path[] inputPaths = new Path[paths.length];
		for (int i = 0; i < paths.length; i++){
			inputPaths[i] = new Path(paths[i]);
		}
		inputPathsList.add(inputPaths);
		inputFormatClassesList.add(inputFormat);
		pathPatternList.add(pathPattern);
		charsetList.add(charset);
		separatorList.add(separator);
		segCountList.add(segCount);
	}
	/**
	 * Add a source table for the current operation.
	 * Calling this method is equivalent to calling addSourceTable(paths, pathPattern, charset, separator, 0).
	 * @param paths all paths that the source table consists of.
	 * @param pathPattern the longest common prefix or longest common substring of paths, or any string that fits. See the manual.
	 * @param charset the character set of the source table, e.g. "UTF-8".
	 * @param separator the separator of the source table, a plain text string, no any wildcard or regular expression supported.
	 */
	public void addSourceTable(String[] paths, String pathPattern, String charset, String separator){
		addSourceTable(paths, TextInputFormat.class, pathPattern, charset, separator, 0);
	}
	
	@SuppressWarnings("unchecked")
	public void addSourceTable(String[] paths, Class<? extends FileInputFormat> inputFormat, String pathPattern, String charset, String separator){
		addSourceTable(paths, inputFormat, pathPattern, charset, separator, 0);
	}
	/*public void addSourceTable(String[] paths, String pathPattern, String charset){
		addSourceTable(paths, pathPattern, charset, null, 0);
	}
	public void addSourceTable(String[] paths, String pathPattern){
		addSourceTable(paths, pathPattern, null, null, 0);
	}*/
	/**
	 * Set the path of the result table, i.e. the output path.
	 * @param path the path of the result table.
	 */
	public void setResultTable(String path){
		this.outputPath = new Path(path);
	}
	/**
	 * Set the separator of the result table.
	 * @param separator the separator of the result table, one or more characters.
	 */
	public void setOutputSeparator(String separator){
		this.outputSeparator = separator;
	}
	/**
	 * Set the representation of a null value in the result table.
	 * @param notation a null field value will be printed as <b><i>notation</i></b>.
	 */
	public void setNullValueNotation(String notation){
		this.nullValueNotation = notation;
	}
	/**
	 * Set a source record filter to filter input records.
	 * @param sourceRecordFilterClass a class that implements the SourceRecordFiler interface.
	 */
	public void setSourceRecordFilterClass(Class<? extends SourceRecordFilter> sourceRecordFilterClass){
		this.sourceRecordFilterClass = sourceRecordFilterClass;
	}
	/**
	 * Set a output record filter to filter output records.
	 * @param outputRecordFilterClass a class that implements the OutputRecordFilter interface.
	 */
	public void setOutputRecordFilterClass(Class<? extends OutputRecordFilter> outputRecordFilterClass){
		this.outputRecordFilterClass = outputRecordFilterClass;
	}
	
	/**
	 * Add a key group for the current operation. See the manual.
	 * @param keyGroup is in the format like "1.1, 2.3, 3.2"
	 */
	public void addKeyGroup(String keyGroup){
		String[] seg = keyGroup.split("[\\s]*[\\s,][\\s]*");
		FieldIndexPair[] fieldIndices = new FieldIndexPair[seg.length];
		HashSet<Integer> sourceIndexSet = new HashSet<Integer>();
		int sourceCount = 0;
		for (int i = 0; i < seg.length; i++){
			String[] part = seg[i].split("\\.");
			if (part.length != 2){
				throw new IllegalArgumentException("wrong format of key-group string: " + keyGroup);
			}
			int sourceIndex = Integer.parseInt(part[0]);
			int columnIndex = Integer.parseInt(part[1]);
			if (sourceIndex <= 0){
				throw new IllegalArgumentException("source index must be positive");
			}
			if (sourceIndexSet.contains(sourceIndex)){
				throw new IllegalArgumentException("one source index cannot appear more than once in a key group");
			}else{
				sourceIndexSet.add(sourceIndex);
			}
			
			fieldIndices[i] = new FieldIndexPair(sourceIndex, columnIndex);
			if (sourceIndex > sourceCount){
				sourceCount = sourceIndex;
			}
		}
		if (fieldIndices.length != sourceCount ||
				sourceCount != inputPathsList.size()){
			throw new IllegalArgumentException("a key group should contain one column from each source table");
		}
		if (keyColumnList == null){
			keyColumnList = new String[sourceCount];
		}else if (keyColumnList.length != sourceCount){
			throw new IllegalArgumentException("source-count not match among key groups");
		}
		for (FieldIndexPair pair : fieldIndices){
			int j = pair.getSourceIndex() - 1;
			int k = pair.getColumnIndex();
			if (keyColumnList[j] == null){
				keyColumnList[j] = new Integer(k).toString();
			}else{
				keyColumnList[j] += " " + new Integer(k).toString();
			}
		}
	}
	
	protected void setUsefulFields(int sourceCount, FieldIndexPair[] fieldIndices){
		valueColumnList = new String[sourceCount];
		for (FieldIndexPair pair : fieldIndices){
			int j = pair.getSourceIndex() - 1;
			int k = pair.getColumnIndex();
			if (valueColumnList[j] == null){
				valueColumnList[j] = new Integer(k).toString();
			}else{
				valueColumnList[j] += " " + new Integer(k).toString();
			}
		}
	}
	
	/**
	 * Set the result fields indices for the current operation.
	 * See the manual.
	 * @param resultFields is in the format like "1.1, 1.2, 2.2, 2.3"
	 */
	public void setResultFields(String resultFields){
		BaseMapredUtils.FieldIndicesStringParseResult res = 
			BaseMapredUtils.parseFieldIndicesString(resultFields);
		int sourceCount = res.getGuessedSourceCount();
		sourceCount = inputPathsList.size();
		FieldIndexPair[] fieldIndices = res.getFieldIndices();
		//make value-column list
		setUsefulFields(sourceCount, fieldIndices);
		//make result-fields-indices parameter for BaseReducer
		for (FieldIndexPair pair : fieldIndices){
			int j = pair.getSourceIndex();
			int k = pair.getColumnIndex();
			String indexStr = new Integer(j).toString() + "." + new Integer(k).toString();
			if (resultFieldsIndices == null){
				resultFieldsIndices = indexStr;
			}else{
				resultFieldsIndices += " " + indexStr;
			}
		}
	}
	/**
	 * Set a {@link RecordGroupFilter} for the job.
	 * A RecordGroupFilter is applied in the reduce phase of a job.
	 * Default filter is always available for common "Runner" classes.
	 * Call this method only if you want to customize the reduce phase operations.
	 * 
	 * @param recordGroupFilterClass a class that implements the {@link RecordGroupFilter} interface
	 * @param usefulColumnFields field indices string like "1.2 2.3 2.4 3.2", where these fields are
	 * useful in the reduce phase. Other fields are ignored in the map phase so that you will get a null
	 * if you read a ignored field.
	 */
	public void setRecordGroupFilterClass(
			Class<? extends RecordGroupFilter> recordGroupFilterClass,
			String usefulColumnFields){
		this.recordGroupFilterClass = recordGroupFilterClass;
		//make value-column list
		BaseMapredUtils.FieldIndicesStringParseResult res = 
			BaseMapredUtils.parseFieldIndicesString(usefulColumnFields);
		int sourceCount = res.getGuessedSourceCount();
		sourceCount = inputPathsList.size();
		FieldIndexPair[] fieldIndices = res.getFieldIndices();
		setUsefulFields(sourceCount, fieldIndices);
	}
	
	/**
	 * Set the number of map tasks.
	 * @param number a positive value set the number of map tasks to the value,
	 * a zero value remains the default number of map tasks.
	 * @throws IllegalArgumentException when number is negative.
	 */
	public void setNumberOfMapTasks(int number){
		if (number < 0){
			throw new IllegalArgumentException("number of map tasks cannot be negative");
		}
		this.numMapTasks = number;
	}
	/**
	 * Set the number of reduce tasks.
	 * @param number a positive value set the number of reduce tasks to the value,
	 * a zero value remains the default number of reduce tasks.
	 * @throws IllegalArgumentException when number is negative.
	 */
	public void setNumberOfReduceTasks(int number){
		if (number < 0){
			throw new IllegalArgumentException("number of reduce tasks cannot be negative");
		}
		this.numReduceTasks = number;
	}
	/**
	 * Set the number of tasks per JVM.
	 * @param number a positive value. Set the number of tasks to the value,
	 * a zero value keeps the default number of tasks.
	 * @throws IllegalArgumentException when number is negative.
	 */
	public void setNumberOfTasksPerJVM(int number){
		if (number < 0){
			throw new IllegalArgumentException("number of tasks per JVM cannot be negative");
		}
		this.numTasksPerJvm = number;
	}
	/**
	 * Set a {@link JobConfigurator} instance, which will be called during execution of {@link #run} method.
	 * @param configurator instance of a class which implements {@link JobConfigurator} interface.
	 */
	public void setJobConfigurator(JobConfigurator configurator){
		this.jobConfigurator = configurator;
	}
	/**
	 * Run the job with a specified job name.
	 * @param jobName the name of the job
	 * @throws IOException
	 */
	public void run(String jobName) throws IOException{
		JobConf conf = new JobConf(BaseMapredRunner.class);
		conf.setJobName(jobName);
		configureJob(conf);
		FileSystem fs = FileSystem.get(conf);
		fs.delete(FileOutputFormat.getOutputPath(conf), true);
		System.out.println("job<" + jobName + "> starts running.");
		JobClient.runJob(conf);
		System.out.println("job<" + jobName + "> completed.");
	}
	
	protected void configureJob(JobConf conf){
		configureJob(conf, true);
	}
	
	@SuppressWarnings("unchecked")
	protected void configureJob(JobConf conf, boolean hasReducer) throws IllegalArgumentException{

		if (inputPathsList.size() != inputFormatClassesList.size())
			throw new IllegalArgumentException("the input format does not match to input path");
//		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		//parameterized types
		Type t = this.getClass();
		while (true){
			if (ParameterizedType.class.isInstance(t)){
				if (((ParameterizedType)t).getRawType().equals(BaseMapredRunner.class)){
					break;
				}
				t = ((Class)((ParameterizedType)t).getRawType()).getGenericSuperclass();
			}else{
				t = ((Class)t).getGenericSuperclass();
			}
		}
		Type[] types1 = ((ParameterizedType)t).getActualTypeArguments();
		//Type[] types1 = ((ParameterizedType)this.getClass().getGenericSuperclass()).getActualTypeArguments();
		Class<? extends Mapper> mapperClass = (Class<? extends Mapper>)types1[0];
		Type[] types2 = ((ParameterizedType)mapperClass.getGenericSuperclass()).getActualTypeArguments();
		Class<?> keyClass = (Class<?>)types2[1];
		Class<?> valueClass = (Class<?>)types2[2];
		conf.setMapperClass(mapperClass);
		conf.setMapOutputKeyClass(keyClass);
		conf.setMapOutputValueClass(valueClass);
		if (hasReducer){
			Class<? extends Reducer> reducerClass;
			reducerClass = (Class<? extends Reducer>)types1[1];
			conf.setReducerClass(reducerClass);
		}
		
		conf.setOutputKeyClass(NullWritable.class);
		conf.setOutputValueClass(Text.class);
	
		//input/output paths
		for (int i = 0; i < inputPathsList.size(); ++i) {
			Path[] paths = inputPathsList.get(i);
			for (Path path : paths){
				
				MultipleInputs.addInputPath(conf, path, inputFormatClassesList.get(i));
			}
		}
		
		FileOutputFormat.setOutputPath(conf, this.outputPath);
		
		//parameters for BaseMapper/BaseReducer
		int sourceCount = inputPathsList.size();
		conf.setInt(BaseMapredParams.SOURCE_COUNT_PARAM_NAME, sourceCount);
		conf.setInt(BaseMapredParams.MAX_REDUCE_ITEM_COUNT_PARAM_NAME, maxReduceItemCount);
		BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.CHARSETS_PARAM_NAME, charsetList);
		BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.SEPARATORS_PARAM_NAME, separatorList);
		BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.SEGCOUNT_LIST_PARAM_NAME, segCountList);
		BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.SOURCE_PATH_PATTERNS_PARAM_NAME, pathPatternList);
		if (keyColumnList != null){
			BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.KEY_COLUMN_LISTS_PARAM_NAME, keyColumnList);
		}
		BaseMapredUtils.setConfigStrings(conf, BaseMapredParams.VALUE_COLUMN_LISTS_PARAM_NAME, valueColumnList);
		if (resultFieldsIndices != null){
			conf.set(BaseMapredParams.RESULT_FIELDS_INDICES_PARAM_NAME, resultFieldsIndices);
		}
		if (sourceRecordFilterClass != null){
			conf.set(BaseMapredParams.SOURCERECORDFILTER_CLASSNAME_PARAM_NAME, sourceRecordFilterClass.getName());
		}
		if (outputRecordFilterClass != null){
			conf.set(BaseMapredParams.OUTPUTRECORDFILTER_CLASSNAME_PARAM_NAME, outputRecordFilterClass.getName());
		}
		if (recordGroupFilterClass != null){
			conf.set(BaseMapredParams.RECORDGROUPFILTER_CLASSNAME_PARAM_NAME, recordGroupFilterClass.getName());
		}
		if (this.outputSeparator != null){
			BaseMapredUtils.setConfigString(conf, 
					BaseMapredParams.SEPARATOR_PARAM_NAME, this.outputSeparator);
		}
		if (this.nullValueNotation != null){
			BaseMapredUtils.setConfigString(conf,
					BaseMapredParams.NULLNOTATION_PARAM_NAME, this.nullValueNotation);
		}
		
		//task configuration
		if (this.numMapTasks > 0) conf.setNumMapTasks(this.numMapTasks);
		if (this.numReduceTasks > 0) conf.setNumReduceTasks(this.numReduceTasks);
		if (this.numTasksPerJvm > 0) conf.setNumTasksToExecutePerJvm(this.numTasksPerJvm);
		
		//customized job configuration
		if (this.jobConfigurator != null){
			this.jobConfigurator.configure(conf);
		}
	}
}
